package zookeeper.client;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import zookeeper.utils.YamlUtil;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 配置中心
 *
 * @author zhenzicheng
 */
@Slf4j
public class ConfigCenter {
  /**
   * 序列化/反序列化
   */
  public static final ObjectMapper objectMapper = new ObjectMapper();
  /**
   * Zookeeper服务所在连接路径 暂时没有部署服务端
   */
  // private static final String CONNECT_STR = "172.21.152.55:2181";
  /**
   * session超时时间 30s
   */
  private final static Integer SESSION_TIMEOUT = 30 * 1000;
  private static ZooKeeper zookeeper = null;
  /**
   * 计数器 初始1
   */
  private static CountDownLatch countDownLatch = new CountDownLatch(1);

  public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
    /*
     * 前置知识点：每次获取监听的数据后，监听器会被server移除，需要再次添加watcher
     */
    zookeeper = new ZooKeeper(getConnectStr(), SESSION_TIMEOUT, watchedEvent -> {
      // 初始化连接后会使用None类型连接作为检测是否连接成功
      if (watchedEvent.getType() == Watcher.Event.EventType.None
          && watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected) {
        log.info("连接已建立");
        // 连接建立好后让主线程继续执行
        countDownLatch.countDown();
      }
    });
    // 等待计数器恢复0
    countDownLatch.await();

    MyConfig myConfig = new MyConfig();
    myConfig.setKey("anyKey");
    myConfig.setName("anyName");
    // 序列化
    byte[] bytes = objectMapper.writeValueAsBytes(myConfig);
    /*
     * 同步创建节点 所有权限 持久化节点
     */
    String path = zookeeper.create("/myconfig", bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    // 创建监听
    Watcher watcher = new Watcher() {
      @SneakyThrows
      @Override
      public void process(WatchedEvent watchedEvent) {
        // 只对NodeDataChange节点监听路径为/myconfig的路径进行事件监听
        if (watchedEvent.getType() == Watcher.Event.EventType.NodeDataChanged
            && watchedEvent.getPath() != null
            && "/myconfig".equals(watchedEvent.getPath())) {
          log.info("PATH:{} 发生了数据变化", watchedEvent.getPath());
          // 获取数据并且再次对当前节点进行监听
          byte[] data = zookeeper.getData("/myconfig", this, null);
          MyConfig newConfig = objectMapper.readValue(new String(data), MyConfig.class);

          log.info("数据发生变化：{}", newConfig);
        }
      }
    };
    // 获取zookeeper节点并添加监听器 状态为null
    byte[] data = zookeeper.getData("/myconfig", watcher, null);
    MyConfig originalMyConfig = objectMapper.readValue(new String(data), MyConfig.class);
    log.info("原始数据：{}", originalMyConfig);
    /*
     * Zookeeper底层创建了两个线程，一个是跟服务端交互的发送线程，一个是监听事件线程
     * 如果主线程运行结束会导致Zookeeper也停止，所以在这里将主线程等待
     */
    TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);

  }

  public static String getConnectStr() {
    return YamlUtil.getDefaultConnectStr();
  }

}
